use std::time;
use std::{str::FromStr, sync::Arc};

use aspect_db_sql_migration::OnConflict;
use futures::StreamExt;
use sea_orm::prelude::*;
use sea_orm::ActiveValue;
use tracing::*;

use aspect_core::message_queue::{MessageQueueId, QueueEntry};
use aspect_db_sql::{device_queue_sub, message_queue};
use aspect_ident::DeviceIdent;
use aspect_msgqueue::Msg;

use crate::async_mq::{AsyncMqDatastore, QueueId, ReaderId};
use crate::errors::*;

#[derive(Clone, Debug)]
pub struct QueueStatus {
    /// Contextual ID of the queue.
    pub id: MessageQueueId,

    /// Position of the last message the reader read.
    pub reader_pos: Option<u64>,

    /// The next message that will be written to the queue.
    pub next_message_pos: u64,
}

pub struct DeviceMqDatastore {
    /// Underlying wrapper around the higher level interface to the MQ data.
    msg_queue: Arc<AsyncMqDatastore>,

    /// Database connection for the high level device data.
    dbc: DatabaseConnection,
}

impl DeviceMqDatastore {
    pub fn new(msg_queue: Arc<AsyncMqDatastore>, dbc: DatabaseConnection) -> Result<Self, Error> {
        Ok(Self { msg_queue, dbc })
    }

    /// Returns the low-level queue ID for a named queue.
    async fn get_queue_id_for_mqid(&self, mqid: &MessageQueueId) -> Result<Option<QueueId>, Error> {
        let queue_m = message_queue::Entity::find()
            .filter(message_queue::Column::Name.eq(mqid.to_string()))
            .one(&self.dbc)
            .await?;

        let Some(queue_m) = queue_m else {
            return Ok(None);
        };

        Ok(Some(QueueId::try_from(queue_m.queue_id.as_ref())?))
    }

    /// Returns the queue ID and reader ID for a device in a named queue.
    async fn get_queue_ids_for_device_queue(
        &self,
        device: &DeviceIdent,
        mqid: &MessageQueueId,
    ) -> Result<Option<(QueueId, ReaderId)>, Error> {
        let sub_m = device_queue_sub::Entity::find()
            .filter(
                device_queue_sub::Column::DeviceId
                    .eq(device.to_vec())
                    .and(device_queue_sub::Column::Name.eq(mqid.to_string())),
            )
            .one(&self.dbc)
            .await?;

        let Some(sub_m) = sub_m else {
            return Ok(None);
        };

        let qid = QueueId::try_from(sub_m.queue_id.as_ref())?;
        let rid = ReaderId::try_from(sub_m.reader_id.as_ref())?;

        Ok(Some((qid, rid)))
    }

    /// Gets all of the names of the queues the device has access to and their
    /// read status for the device.
    pub async fn get_device_queues(&self, device: &DeviceIdent) -> Result<Vec<QueueStatus>, Error> {
        let mut queue_ms = device_queue_sub::Entity::find()
            .filter(device_queue_sub::Column::DeviceId.eq(device.to_vec()))
            .stream(&self.dbc)
            .await?;

        let mut qstats = Vec::new();
        while let Some(ent) = queue_ms.next().await.transpose()? {
            let Ok(id) = MessageQueueId::from_str(&ent.name) else {
                warn!(%ent.name, "ignoring unknown queue name");
                continue;
            };

            let qid = QueueId::try_from(ent.queue_id.as_ref())?;
            let rid = ReaderId::try_from(ent.reader_id.as_ref())?;

            let qv = self.msg_queue.query_queue_view(&qid, Some(rid)).await?;

            qstats.push(QueueStatus {
                id,
                reader_pos: qv.last_consumed_idx(),
                next_message_pos: qv.next_idx(),
            })
        }

        Ok(qstats)
    }

    /// Queries a range of messages in a specific queue, starting at a position
    /// and returning until reaching the end of the range or up until an
    /// optitonal cumulative payload size.
    pub async fn query_device_queue_range(
        &self,
        device: &DeviceIdent,
        mqid: &MessageQueueId,
        start_pos: u64,
        max_len: u64,
        max_payload_size: Option<u64>,
    ) -> Result<Vec<QueueEntry>, Error> {
        let (qid, rid) = self
            .get_queue_ids_for_device_queue(device, &mqid)
            .await?
            .ok_or_else(|| Error::DeviceMissingQueue(device.clone(), mqid.clone()))?;
        let ret_data_limit = max_payload_size.unwrap_or(u64::MAX);
        let mut cur_payload_size = 0;

        let mut entries = Vec::new();

        let messages = self
            .msg_queue
            .query_queue_message_range(&qid, &rid, start_pos, max_len)
            .await?;

        for (idx, mdata) in messages {
            let len = mdata.data().len() as u64;
            if cur_payload_size + len > ret_data_limit {
                break;
            }

            cur_payload_size += len;

            let ent = QueueEntry {
                idx,
                timestamp: mdata.timestamp(),
                data: mdata.into_data(),
            };

            entries.push(ent);
        }

        Ok(entries)
    }

    /// Returns if a device has access to a queue with some MQID.
    pub async fn check_device_queue_access(
        &self,
        device: &DeviceIdent,
        mqid: &MessageQueueId,
    ) -> Result<bool, Error> {
        let cnt = device_queue_sub::Entity::find()
            .filter(
                device_queue_sub::Column::DeviceId
                    .eq(device.to_vec())
                    .and(device_queue_sub::Column::Name.eq(mqid.to_string())),
            )
            .count(&self.dbc)
            .await?;
        Ok(cnt > 0)
    }

    /// Simply returns the next message index of the queue.
    pub async fn get_queue_next_idx(
        &self,
        device: &DeviceIdent,
        mqid: &MessageQueueId,
    ) -> Result<u64, Error> {
        let (qid, rid) = self
            .get_queue_ids_for_device_queue(device, mqid)
            .await?
            .ok_or_else(|| Error::DeviceMissingQueue(device.clone(), mqid.clone()))?;
        let qv = self.msg_queue.query_queue_view(&qid, Some(rid)).await?;
        Ok(qv.next_idx())
    }

    /// Submits a message to a queue.
    pub async fn submit_msg_to_queue(
        &self,
        mqid: &MessageQueueId,
        msg: &[u8],
    ) -> Result<u64, Error> {
        trace!(%mqid, len = %msg.len(), "submitting msg to queue");

        // Look up the queue ID.
        let mqid_name = mqid.to_string();
        let mq_m = message_queue::Entity::find()
            .filter(message_queue::Column::Name.eq(mqid_name))
            .one(&self.dbc)
            .await?
            .ok_or_else(|| Error::UnknownMessageQueue(mqid.clone()))?;
        let qid = QueueId::try_from(mq_m.queue_id.as_ref())?;
        trace!(%qid, "found low-level queue");

        // Then construct a message and insert it.
        let msg = Msg::new(now_ts(), msg.to_vec());
        let idx = self.msg_queue.insert_msg_into_queue(&qid, msg).await?;

        Ok(idx)
    }

    /// Inserts multiple messages into a queue from an iterator.
    pub async fn submit_msgs_to_queue<'m, I: Iterator<Item = &'m [u8]>>(
        &self,
        mqid: &MessageQueueId,
        msgs_iter: I,
    ) -> Result<Vec<u64>, Error> {
        // Look up the queue ID.
        let mqid_name = mqid.to_string();
        let mq_m = message_queue::Entity::find_by_id(mqid_name)
            .one(&self.dbc)
            .await?
            .ok_or_else(|| Error::UnknownMessageQueue(mqid.clone()))?;
        let qid = QueueId::try_from(mq_m.queue_id.as_ref())?;

        // Insert each of them now that we have the queue ID.
        let mut idxs = Vec::new();
        for m in msgs_iter {
            let msg = Msg::new(now_ts(), m.to_vec());
            let idx = self.msg_queue.insert_msg_into_queue(&qid, msg).await?;
            idxs.push(idx);
        }

        Ok(idxs)
    }

    /// Tries to subscribe a device to a queue, returning if we newly subscribed
    /// them.
    pub async fn sub_device_to_queue(
        &self,
        mqid: &MessageQueueId,
        device: &DeviceIdent,
    ) -> Result<bool, Error> {
        // TODO this is kinda maybe a race, rework all this so it happens as a
        // single transaction and most of the work happens on the server

        // Check that the device isn't already registered.
        if self
            .get_queue_ids_for_device_queue(device, mqid)
            .await?
            .is_some()
        {
            return Ok(false);
        }

        // Get the queue ID and add a new reader.
        let rid = ReaderId::new_rand();
        let qid = self
            .get_queue_id_for_mqid(mqid)
            .await?
            .ok_or_else(|| Error::UnknownMessageQueue(mqid.clone()))?;

        // Insert the subscription into the table.  We do this first so that if
        // we crash before Sled flushes to disk we can detect the dangling
        // reader instead of having a resource leak.
        let sub_am = device_queue_sub::ActiveModel {
            device_id: ActiveValue::Set(aspect_codec::encode_to_vec(device)?),
            name: ActiveValue::Set(mqid.to_string()),
            queue_id: ActiveValue::Set(qid.to_vec()),
            reader_id: ActiveValue::Set(rid.to_vec()),
            ..Default::default()
        };

        let rows = device_queue_sub::Entity::insert(sub_am)
            .exec_without_returning(&self.dbc)
            .await?;

        if rows != 1 {
            error!(%mqid, %device, "unable to insert device subscription entry, something is wrong");
            return Err(Error::UnableToInsertSub(*device, mqid.clone()));
        }

        // Now that we've inserted the DB entry we can actually add the reader,
        // this should always work.
        let succ = self.msg_queue.subscribe_reader_to_queue(&qid, &rid).await?;
        assert!(
            succ,
            "device_msg_queue: fresh reader ID already present in queue???"
        );

        Ok(true)
    }

    /// Unsubscribes a device from a queue.  If they're already unsubscribed, does nothing.
    ///
    /// If there's an inconsistency between the subscriptions table and the
    /// low-level reader table, will clean up dangling reader IDs.
    pub async fn unsub_device_from_queue(
        &self,
        mqid: &MessageQueueId,
        device: &DeviceIdent,
    ) -> Result<bool, Error> {
        let Some((qid, rid)) = self.get_queue_ids_for_device_queue(device, mqid).await? else {
            return Ok(false);
        };

        let succ = self
            .msg_queue
            .unsubscribe_reader_from_queue(&qid, &rid)
            .await?;
        if !succ {
            warn!(%mqid, %device, "tried to remove device reader from MQ but it was already missing, did we crash recently?");
        }

        let res = device_queue_sub::Entity::delete_many()
            .filter(
                device_queue_sub::Column::Name
                    .eq(mqid.to_string())
                    .and(device_queue_sub::Column::DeviceId.eq(device.to_vec())),
            )
            .exec(&self.dbc)
            .await?;

        let rows = res.rows_affected;

        if rows == 0 {
            warn!(%mqid, %device, "removing device from subs table yielded 0 rows, wtf?");
        }

        if rows > 1 {
            warn!(%mqid, %device, %rows, "removing device from subs table yielded multiple rows, wtf?");
        }

        Ok(true)
    }

    /// Creates a uniquely-named queue and the corresponding low level queue.
    /// Returns if it was successful.
    pub async fn create_queue(&self, mqid: MessageQueueId) -> Result<bool, Error> {
        // It's safer to have empty dangling queues so let's create that first.
        // We still risk it not being flushed by the time we finish the SQL
        // insert but that still makes it less likely to screw something up.

        // Don't add any initial readers.
        let llqid = self.msg_queue.create_mq(Vec::new()).await?;

        let queue_am = message_queue::ActiveModel {
            name: ActiveValue::Set(mqid.to_string()),
            queue_id: ActiveValue::Set(llqid.to_vec()),
        };

        let res = message_queue::Entity::insert(queue_am)
            .on_conflict(OnConflict::new().do_nothing().to_owned())
            .exec_without_returning(&self.dbc)
            .await?;

        if res == 0 {
            // TODO go back and delete the low-level queue we just created
            warn!(%mqid, %llqid, "tried to create duplicate queue, doing nothing");
            return Ok(false);
        }

        trace!(%mqid, %llqid, "created queue");

        Ok(true)
    }
}

fn now_ts() -> u64 {
    time::UNIX_EPOCH.elapsed().expect("db: now_ts").as_millis() as u64
}
